package com.dx.logic.transfer;

import com.dx.cache.cache.DmpRedisCache;
import com.dx.chat.kafka.config.service.AbstractKafkaConsumerService;
import com.dx.logic.PushMsgBoxService;
import com.dx.mobileimsdk.server.protocal.Protocal;
import com.dx.mobileimsdk.server.utils.ServerToolKits;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Map;

/**
 * kafka transfer
 *
 * @author xufeng
 * @since 13.04.2023
 */
@Service
@Slf4j
public class KafkaConsumerService extends AbstractKafkaConsumerService {

    @Resource
    DmpRedisCache dmpRedisCache;
    @DubboReference
    private PushMsgBoxService pushMsgBoxService;

    /**
     * 用户发送消息  持久化到数据库等逻辑
     * logic 模块监听
     * 服务器到服务器
     *
     * @param record s2s_chat = 发件箱
     */
    @KafkaListener(topics = "s2s_chat", containerFactory = "dataGroupKafkaListenerContainerFactory")
    @Override
    public void dataPersistenceListener(ConsumerRecord<String, String> record) {
        String value = record.value();
        log.info("【消息发件箱->mysql】:{}", value);
    }

    /**
     * 用户发送离线消息，持久化到mongodb
     * logic 模块监听
     * 服务器到 phsh
     *
     * @param record
     */
    @KafkaListener(topics = "s2s_chat", containerFactory = "dataGroupKafkaListenerContainerFactory2")
    @Override
    public void offlineMsgListener(ConsumerRecord<String, String> record) {
        String value = record.value();
        log.info("【消息发件箱->mongo DB】:{}", value);
    }

    /**
     * Pusher 模块监听
     * s2m_chat = 收件箱
     *
     * @param record
     */
    @KafkaListener(topics = "s2m_chat", containerFactory = "dataGroupKafkaListenerContainerFactoryPusher")
    @Override
    public void listenerPusher(ConsumerRecord<String, String> record) {
        String value = record.value();
        Protocal protocal = null;
        try {
            protocal = ServerToolKits.toProtocal(value);
        } catch (Exception e) {
            log.info("序列化用户信息发生异常");
            throw new RuntimeException(e);
        }
        Map<Object, Object> map = dmpRedisCache.hGetAll(protocal.getTo());
        if (map.size() == 0) {
            log.info("用户不在线，存入monGodb，已在其它收件箱处理。这里不消费即可.");
            return;
        }
        log.info("【消息收件箱->grpc to pusher】:{},{}", value, new Gson().toJson(map));
        pushMsgBoxService.receiveMsg(value, String.valueOf(map.get("ip")), String.valueOf(map.get("dubboPort")));

    }
}
